#include <stdio.h>
#include <stdlib.h>
#include <signal.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <sys/socket.h>

#include <mutex>
#include <atomic>
#include <thread>

#include <vector>
#include <set>

#include "common/net_utils.h"
#include "net_tools.h"

#ifndef USE_THREAD
#include "co_api.h"
#endif

using namespace std;

#define SERVER_HOST "127.0.0.1"
#define SERVER_PORT 6666

#define MAX_FD 1000000

int flow_test(int port);

int do_accept_worker(int workers, int port);

int main(int argc, char** argv)
{
   // LogDefaulter::set_log_level(LOG_LEVEL_DEBUG);
    LogDefaulter::set_log_level(LOG_LEVEL_INFO);
   // LogDefaulter::set_log_level(LOG_LEVEL_ERROR);
    LogDefaulter::set_logger([](const string& message) {
        printf("socket_test|%s\n", message.c_str());         
		fflush(stdout);
    });

    NetTools::ignore_sig_pipe();

    int port = SERVER_PORT; 
    if (argc >= 2) {
        port = atoi(argv[1]);
    }

    flow_test(port);
    PRINTF("main end");

    return 0;
}

int do_accept_worker(int workers, int port, mutex& mu, set<int>& client_fds, CoServerCb& cb)
{
	for (int i = 0; i < workers; i++) {

		thread([port, &mu, &client_fds, cb] {

	        int backlog = 512;

	        int s = socket(AF_INET, SOCK_STREAM, 0);
	        assert(s >= 0);

	        int ret = NetUtils::set_socket_reuseport(s);
	        assert(ret == 0);
	        PRINTF("set_socket_reuseport succeed, fd:%d", s);

	        ret = NetUtils::set_socket_reuseaddr(s);	
	        assert(ret == 0);
	        PRINTF("set_socket_reuseaddr succeed, fd:%d", s);

	        ret = NetUtils::bind_and_listen(s, "", port, backlog);
            if (ret != 0) {
                PRINTF("bind_and_listen fail, fd:%d, errno:%d", s, errno);
            }
	        assert(ret == 0);
	        PRINTF("bind_and_listen succeed, fd:%d", s);

            while (1) {

			    default_random_engine e;
        	    uniform_int_distribution<int> u(0, 1000000);
        	    e.seed(time(0));

        	    int seq = u(e);

			    struct sockaddr_in caddr;
			    socklen_t len = sizeof(caddr);

			    int c = accept(s, (struct sockaddr*)&caddr, &len);
			    if (c < 0) {
			    	PRINTF("ERROR|accept fail, seq:%d, accept_fd:%d, errno:%d", seq, s, errno);
			    	exit(0);
			    }

			    if (NetUtils::set_tcp_no_delay(c) != 0)
			    {
			    	PRINTF("ERROR|set_tcp_no_delay fail, seq:%d, client_fd:%d");
			    	exit(0);
			    }

			    NetTools::create_task([c, cb] {
			    	NetTools::echo_receiver(c, cb);
			    });

			    lock_guard<mutex> lock(mu);
			    client_fds.insert(c);	
            }

		}).detach();
	}

	return 0;
}

int flow_test(int port)
{
    long last_counters = 0;

	atomic<long> total_send_cost_us(0);
	atomic<long> total_recv_cost_us(0);

	set<int> client_fds;
	vector<long>          fd_recv_old_counters;
	vector<atomic<long>*> fd_recv_now_counters;
    vector<atomic<long>*> counters;

	mutex mu;

    auto co_env_keys  = CoApi::get_env_keys();
    auto co_env_param = CoApi::get_env_param();

    PRINTF("listen.port:%d", port);

    for (int i = 0; i < co_env_param.schedule_count; i++) {
        counters.push_back(new atomic<long>(0));
    }
	for (int i = 0; i < MAX_FD; i++) {
		fd_recv_old_counters.push_back(0);
		fd_recv_now_counters.push_back(new atomic<long>(0));
	}

	CoServerCb cb;
	cb.send_cb = [&total_send_cost_us](int fd, const string& buffer, bool is_ok, long cost_us) {
		if (is_ok) {
			total_send_cost_us += cost_us;
		}
	};
	cb.recv_cb = [&mu, &client_fds, &counters, &total_recv_cost_us, &fd_recv_now_counters](
			int fd, 
			const string& buffer, 
			bool is_ok, 
			long cost_us)
	{
        int sid = CoApi::get_schedule_id();
		if (is_ok) {
			total_recv_cost_us += cost_us;
			(*fd_recv_now_counters[fd])++;
            (*counters[sid])++;
		} else {
			lock_guard<mutex> lock(mu);
			client_fds.erase(fd);
		}
	};
	cb.close_cb = [&client_fds, &mu](int fd) {
		lock_guard<mutex> lock(mu);
		client_fds.erase(fd);
	};

	do_accept_worker(8, port, mu, client_fds, cb);

	long last_total_send_cost_us = total_send_cost_us.load();
	long last_total_recv_cost_us = total_recv_cost_us.load();

	auto last_co_statics = CoApi::get_statics();

    for (const auto& item : co_env_keys) {
        PRINTF("ENV_KEY: %s", item.c_str());
    }
    PRINTF("");

    while (1) {
        sleep(1);
        long now_counters = 0;
        long delta_counters;
		long delta_total_send_cost_us;
		long delta_total_recv_cost_us;
		{
			long temp_total_send_cost_us = total_send_cost_us.load();
			long temp_total_recv_cost_us = total_recv_cost_us.load();

            for (auto& item : counters) {
                now_counters += item->load();
            }

			delta_counters = now_counters - last_counters;
			delta_total_send_cost_us = temp_total_send_cost_us - last_total_send_cost_us;
			delta_total_recv_cost_us = temp_total_recv_cost_us - last_total_recv_cost_us;

			last_counters = now_counters;
			last_total_send_cost_us = temp_total_send_cost_us;
			last_total_recv_cost_us = temp_total_recv_cost_us;
		}
		auto co_statics = CoApi::get_statics();	
		const auto& reactor_counter = co_statics.reactor_counter;

		double send_rtt = delta_counters > 0 ? ((double)delta_total_send_cost_us / delta_counters) / 1000 : 0;
		double recv_rtt = delta_counters > 0 ? ((double)delta_total_recv_cost_us / delta_counters) / 1000 : 0;

		long delta_switch_wait_counter = co_statics.total_switch_wait_counter - last_co_statics.total_switch_wait_counter;
		long delta_switch_wait_cost_counter = co_statics.total_switch_wait_cost_counter - last_co_statics.total_switch_wait_cost_counter;

		double avg_wait_ms = 0.0;
		if (delta_switch_wait_counter > 0) {
			avg_wait_ms = (double)delta_switch_wait_cost_counter / delta_switch_wait_counter;
		}

		int connects;
		{
			lock_guard<mutex> lock(mu);
			connects = (int)client_fds.size();
		}

        PRINTF(
            "########## env_info:{stack_size:%d, co_count:%d, schedule_count:%d, epoll_worker_count:%d}", 
            co_env_param.stack_size,
            co_env_param.co_count,
            co_env_param.schedule_count,
            co_env_param.epoll_worker_count
        );
		PRINTF(
			"++++++++++ qps:%ld, send_rtt:%0.2fms, recv_rtt:%0.2fms, connects:%d", 
			delta_counters, 
			send_rtt,
			recv_rtt,
			connects
		);
		PRINTF(
			"---------- co:{total:%d, free:%d, ready:%d, suspend:%d}",
			co_statics.co_count, 
			co_statics.co_count - co_statics.co_ready_count - co_statics.co_suspend_count,
			co_statics.co_ready_count,
			co_statics.co_suspend_count
		);
		PRINTF(
			"---------- co_switch:{total:%ld, wait:%ld, avg_wait_ms:%0.2fms}", 
			co_statics.total_switch_counter - last_co_statics.total_switch_counter,
			co_statics.total_switch_wait_counter - last_co_statics.total_switch_wait_counter,
			avg_wait_ms
		);

		long delta_epoll_handle_fd_counter = 
			co_statics.reactor_counter.epoll_handle_fd_counter - 
			last_co_statics.reactor_counter.epoll_handle_fd_counter;

		long delta_epoll_wait_counter = 
			co_statics.reactor_counter.epoll_wait_counter - 
			last_co_statics.reactor_counter.epoll_wait_counter;

		long delta_epoll_wait_cost_counter = 
			co_statics.reactor_counter.epoll_wait_cost_counter - 
			last_co_statics.reactor_counter.epoll_wait_cost_counter;

		long delta_epoll_mod_counter = 
			co_statics.reactor_counter.epoll_mod_counter - 
			last_co_statics.reactor_counter.epoll_mod_counter;

		long delta_epoll_mod_cost_counter = 
			co_statics.reactor_counter.epoll_mod_cost_counter - 
			last_co_statics.reactor_counter.epoll_mod_cost_counter;

		long delta_epoll_add_counter = 
			co_statics.reactor_counter.epoll_add_counter - 
			last_co_statics.reactor_counter.epoll_add_counter;

		long delta_epoll_add_cost_counter = 
			co_statics.reactor_counter.epoll_add_cost_counter - 
			last_co_statics.reactor_counter.epoll_add_cost_counter;

		long delta_epoll_del_counter = 
			co_statics.reactor_counter.epoll_del_counter - 
			last_co_statics.reactor_counter.epoll_del_counter;

		long delta_epoll_del_cost_counter = 
			co_statics.reactor_counter.epoll_del_cost_counter - 
			last_co_statics.reactor_counter.epoll_del_cost_counter;


		double avg_epoll_wait_cost = 0.0;
		double avg_epoll_handle_fd_counter = 0.0;

        double avg_epoll_mod_cost = 0.0;
        double avg_epoll_add_cost = 0.0;
        double avg_epoll_del_cost = 0.0;

		if (delta_epoll_wait_counter) {
			avg_epoll_wait_cost = (double)delta_epoll_wait_cost_counter / delta_epoll_wait_counter;
			avg_epoll_handle_fd_counter = (double)delta_epoll_handle_fd_counter / delta_epoll_wait_counter;
		}
        if (delta_epoll_mod_counter) {
            avg_epoll_mod_cost = (double)delta_epoll_mod_cost_counter / delta_epoll_mod_counter;
        }
        if (delta_epoll_add_counter) {
            avg_epoll_add_cost = (double)delta_epoll_add_cost_counter / delta_epoll_add_counter;
        }
        if (delta_epoll_del_counter) {
            avg_epoll_del_cost = (double)delta_epoll_del_cost_counter / delta_epoll_del_counter;
        }

        PRINTF(
            "---------- co_epoll_wait:{counter:%ld, avg_cost:%0.2fus, handle_fd:%0.2f/per}", 
            delta_epoll_wait_counter, 
            avg_epoll_wait_cost, 
            avg_epoll_handle_fd_counter
        );
        PRINTF("---------- co_epoll_mod:{counter:%ld, avg_cost:%0.2fus}", delta_epoll_mod_counter, avg_epoll_mod_cost);
        PRINTF("---------- co_epoll_add:{counter:%ld, avg_cost:%0.2fus}", delta_epoll_add_counter, avg_epoll_add_cost);
        PRINTF("---------- co_epoll_del:{counter:%ld, avg_cost:%0.2fus}", delta_epoll_del_counter, avg_epoll_del_cost);

		map<int, long> deltas;
		for (int i = 0; i < MAX_FD; i++) {

			long now_counter = fd_recv_now_counters[i]->load();
			long delta = now_counter - fd_recv_old_counters[i];
			if (delta > 0) {
				deltas[i] = delta;
				fd_recv_old_counters[i] = now_counter;
			}
		}
		PRINTF("---------- fd_receiver:{no_recv_fds:%ld}", connects - deltas.size());
		printf("\n");

		last_co_statics = co_statics;
    }

    return 0;
}

